package com.johnguo.beam_test.read;

import java.io.IOException;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.hadoop.inputformat.HadoopInputFormatIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

public class HdfsDemoBeam {
	public static void main(String... args) throws IllegalArgumentException, IOException {
		Configuration conf = new Configuration();
		// FileInputFormat.addInputPath(job, path);
		conf.setClass("mapreduce.job.inputformat.class", TextInputFormat.class, InputFormat.class);
		conf.setClass("key.class", LongWritable.class, Object.class);
		conf.setClass("value.class", Text.class, Object.class);
		conf.set(FileInputFormat.INPUT_DIR, "hdfs://192.168.1.181:9000/domain_history.txt");
		SimpleFunction<LongWritable, String> myKeyTranslate;
		SimpleFunction<Text, String> myValueTranslate;
		myKeyTranslate = new SimpleFunction<LongWritable, String>() {
			@Override
			public String apply(LongWritable input) {
				return input.toString();
			}
		};
		myValueTranslate = new SimpleFunction<Text, String>() {
			@Override
			public String apply(Text input) {
				return input.toString();
			}
		};
		PipelineOptions pipe = PipelineOptionsFactory.create();
		Pipeline p = Pipeline.create(pipe);
		HadoopInputFormatIO.Read<String, String> read = HadoopInputFormatIO.<String, String>read()
				.withConfiguration(conf).withKeyTranslation(myKeyTranslate).withValueTranslation(myValueTranslate);
		p.apply(read).apply("ExtractWords", ParDo.of(new DoFn<KV<String, String>, String>() {
			@ProcessElement
			public void processElement(ProcessContext c) {
				for (String word : c.element().getValue().toString().split(" ")) {
					if (!word.isEmpty()) {
						c.output(word);
					}
				}
			}
		})).apply(Count.<String>perElement()).apply("FormatResult",
				MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
					@Override
					public String apply(KV<String, Long> input) {
						System.out.println(input.getKey() + ": " + input.getValue());
						return input.getKey() + ": " + input.getValue();
					}
				}));
		p.run().waitUntilFinish();
	}
}
